package com.aiguigu.cn.marketanalysis

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: 广告点击统计
  * @Date: 2020/5/6 16:21 
  */
//输入的广告点击事件样例类
case class AdClickEvent( userId: Long, adId: Long, province: String, city: String, timestamp: Long)
//按照省份统计的输出结果样例类
case class  CountByProvince(windowEnd: String, province: String, count: Long)
//输出的黑名单报警信息
case class BlackListWarning( userId: Long, adId: Long, msg: String)

//和地理位置有关广告的点击量统计
object AdStatisticsByGeo {

  //定义测输出流的标签
  val blackListOutputTag: OutputTag[BlackListWarning] = new OutputTag[BlackListWarning]("blackList")

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val source = getClass.getResource("/AdClickLog.csv")
    val adEventStream = environment.readTextFile(source.getPath)
      .map( data => {
        val dataArray = data.split(",")
        AdClickEvent(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim, dataArray(3).trim, dataArray(4).trim.toLong)
      })
      //设置事件戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
      .assignAscendingTimestamps(_.timestamp * 1000L)

    //自定义process function,过滤大量刷点击的行为，并将刷点击行为的用户信息存入侧输出流
    val filterBlackListStream = adEventStream
      .keyBy(data => (data.userId, data.adId))
      //底层API:ProcessFunction ,属于大招，啥都能干
      .process(new FilterBlackListUser(100))


    //根据省份做分组，开窗聚合
    val adCountStream = filterBlackListStream
      .keyBy(_.province)  //要是对多个字段进行分组 就用二元组
      //开窗聚合，滑动窗口
      .timeWindow(Time.hours(1), Time.seconds(5))
      //窗口聚合
      .aggregate(new AdCountAgg(), new AdCountResult() )

    adCountStream.print("ad count")
    //输出刷单行为列表：根据侧输出流获取
    filterBlackListStream.getSideOutput(blackListOutputTag).print()
    environment.execute("ad statistics job")
  }


  //                        --类型[key是二元组(userId,adId), 输入是AdClickEvent, 输出也是AdClickEvent]
  class FilterBlackListUser(maxCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]{

    //定义状态：保存当前用户对当前广告的点击量，前面已经根据用户和广告进行keyBy分组了
    lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count-state", classOf[Long]))
    //保存是否发送过黑名单的状态,避免对同一个人发送多次
    lazy val isSentBlackList: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("issent-state",classOf[Boolean]))
    //保存定时器触发的时间戳
    lazy val resetTimer: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("resettime-state",classOf[Long]))

    //每一条数据进来后的操作
    override def processElement(value: AdClickEvent, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#Context, out: Collector[AdClickEvent]): Unit = {
      //取出count状态
      val curCount = countState.value()

      //如果是第一次处理，注册定时器，每天零点00:00触发
      if (0 == curCount){
        val ts = ( ( ctx.timerService().currentProcessingTime() / (1000*60*60*24) ) + 1 ) * (1000*60*60*24)
        resetTimer.update(ts)
        //要用处理时间processingTime
        ctx.timerService().registerProcessingTimeTimer(ts)
      }

      //判断计数是否达到上限，如果到达则加入黑名单
      if (curCount >= maxCount) {
        //判断是否发送过黑名单，只发送一次
        if(! isSentBlackList.value()){
          isSentBlackList.update(true)
          //输出到测输出流
          ctx.output(blackListOutputTag, BlackListWarning(value.userId, value.adId, "Click over "+ curCount + ">=" + maxCount + "times today."))
        }
        //如果当前用户点击当前广告超过了100次，主流就不再正常统计输出了，而是直接返回
        return
      }

      //计数状态 +1 ，输出到主流
      countState.update(curCount + 1)
      //如果统计的数量没有达到上线，就原封不动的返回出去继续下一步开窗聚合
      out.collect(value)
    }

    //定时器回调函数
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#OnTimerContext, out: Collector[AdClickEvent]): Unit = {
      //定时器触发时，清空状态
      if (timestamp == resetTimer.value()){
        countState.clear()
        isSentBlackList.clear()
        resetTimer.clear()
      }
    }
  }
}

//计数器,自定义预聚合函数
class AdCountAgg() extends AggregateFunction[AdClickEvent, Long, Long]{
  override def createAccumulator(): Long = 0L

  override def add(value: AdClickEvent, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long =  accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

//自定义窗口函数，输出 CountByProvince            ---参数(AdCountAgg()函数的输出, 输出样例类CountByProvince, key为.keyBy(_.province), TimeWindow)
class AdCountResult() extends WindowFunction[Long, CountByProvince, String, TimeWindow]{
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[CountByProvince]): Unit = {
    out.collect(CountByProvince(new Timestamp(window.getEnd).toString, key, input.iterator.next()))
  }
}


